package com.mq.config;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Collections;
import java.util.Properties;

/**
 * <h3>partitionjob</h3>
 * <p>kafka测试生产者消费者</p>
 *
 * @author : hduong
 * @version : 1.0
 * @date : 2019-12-26 09:04
 **/
public class KafkaApp {


    public static String topic = "test";


    public static class MyConsumer {

        KafkaConsumer<String, String> kafkaConsumer;

        public MyConsumer() {
            Properties map = new Properties();
            map.put("bootstrap.servers", "192.168.80.3:9092");
            map.put("group.id", "local-test-1");
            map.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            map.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            kafkaConsumer = new KafkaConsumer<>(map);
            kafkaConsumer.subscribe(Collections.singleton("eagle"));
        }

        public void consumer() {
            System.out.println("wait for consume...");
            try {
                while (true) {
                    ConsumerRecords<String, String> records = kafkaConsumer.poll(100);
                    for (ConsumerRecord<String, String> record : records) {
                        System.out.println("key = " + record.key());
                        System.out.println("value = " + record.value());
                        System.out.println("partition = " + record.partition());
                        System.out.println("topic = " + record.topic());
                        System.out.println("offset = " + record.offset());
                        System.out.println("timestamp = " + record.timestamp());
                        System.out.println();
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        public static void main(String[] args) {
            MyConsumer consumer = new MyConsumer();
            consumer.consumer();
        }
    }

    public static class MyProducer {

        private KafkaProducer<String, String> kafkaProducer;

        public MyProducer() {
            Properties map = new Properties();
            map.put("bootstrap.servers", "192.168.80.3:9092");
            map.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            map.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            kafkaProducer = new KafkaProducer<>(map);
        }

        int i = 0;
        String msg = "bb hh ";

        public void produce() {
            ProducerRecord<String, String> record = new ProducerRecord<String, String>("eagle", msg + i);
            i++;
            try {
                kafkaProducer.send(record, (recordMetadata, e) -> System.out.println("send success"));
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        public static void main(String[] args) {
            MyProducer producer = new MyProducer();
            while (true) {
                producer.produce();
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

}
